package com.erbao.io.netty.refactor.transport;

import com.erbao.IoUtil;
import com.erbao.io.netty.refactor.rpc.ResponseMappingCallBack;
import com.erbao.io.netty.refactor.rpc.protocol.MyContent;
import com.erbao.io.netty.refactor.rpc.protocol.MyHeader;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;

import java.io.*;
import java.net.*;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

public enum ClientFactory{
    INSTANCE;
    int poolSize = 1;
    NioEventLoopGroup clientWorker;

    Random random = new Random();
    /*一个consumer（client）可以连接很多的provider（server），每一个server都有自己的pool
    * key server的ip+port value: */
    ConcurrentHashMap<InetSocketAddress, ClientPool> oubBoxs = new ConcurrentHashMap<>();

    public CompletableFuture<Object> transport (MyContent content) throws InterruptedException {
        // 对象转成字节数组，channle 发送使用 bytebuf。对象进，出字节数组
        byte[] msgBodyBytes = IoUtil.serialize(content);

        // TODO: 23/4/6 未来的小火车可能会变
        /*2.requestId + message ，本地要缓存requestId
         * 自定义协议：【header：请求相关的信息】+【body：msgBody】*/
        MyHeader header = MyHeader.createHeader(msgBodyBytes);
        byte[] msgHeaderBytes = IoUtil.serialize(header);

        // 3.连接池获取连接
        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 9090);
        /*1.缺失了注册返现
         * 2.第一层负载面向provider
         * 3.consumer 线程池 面向service 有并发就会有木桶效应，就是倾斜
         *   serviceA：
         *     ipA：portA
         *       socket1
         *       socket2
         *     ipB: portB
         * 同一个服务注册多个；使用同一个连接（复用连接）给服务发送数据还是使用多个连接发送数据*/
        NioSocketChannel client = getClient(address);

        /*注册响应，即响应后代码继续执行*/
        CompletableFuture<Object> completableFuture = new CompletableFuture<>();
        ResponseMappingCallBack.addCallback(header.getRequestId(), completableFuture);

        // 4.发送 走的IO流 netty（event 驱动）
        System.out.println("发送的header和body的length=="+msgHeaderBytes.length + ":" +msgBodyBytes.length);
        ByteBuf reqBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgHeaderBytes.length + msgBodyBytes.length);
        reqBuf.writeBytes(msgHeaderBytes);
        reqBuf.writeBytes(msgBodyBytes);
        client.writeAndFlush(reqBuf).sync(); //流肯定是双向的，这行代码结束只代表out结束，还需要等待响应，就需要阻塞住

        return completableFuture;
    }

    public CompletableFuture<Object> transport (MyContent content, TransportTypeEnum transportType) throws InterruptedException {
        /*dubbo 是一个rpc框架；netty是一个io框架
        * dubbo 传输协议可以是自定义rpc协议，也可以是http协议*/
        if (TransportTypeEnum.HTTP.equals(transportType)) { // http
            return transportHttp (content);
        } else if (TransportTypeEnum.RPC.equals(transportType)) {
            return transport(content);
        }
        return null;
    }

    private CompletableFuture<Object> transportHttp(MyContent content) {
        /*使用url县城的工具，包含了编解码，发送，socket，连接等
        * on netty。io框架，已经提供的http编解码*/
        CompletableFuture<Object> completableFuture = new CompletableFuture<>();
        // urlHttp (content, completableFuture);
        nettyHttp (content, completableFuture);
        return completableFuture;
    }

    NioEventLoopGroup loopGroup = new NioEventLoopGroup(1);
    Bootstrap bootstrap = new Bootstrap();
    private void nettyHttp(MyContent content, CompletableFuture<Object> completableFuture) {
        /*通过netty建立连接
        * 发送和接收可以异步的形式*/
        Bootstrap client = bootstrap.channel(NioSocketChannel.class)
                .group(loopGroup)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new HttpClientCodec())
                                .addLast(new HttpObjectAggregator(1024 * 512))
                                .addLast(new MyHttpResponseHandler(completableFuture));// 接收数据
                    }
                });

        // 连接
        ChannelFuture channelFuture = null;
        try {
            channelFuture = client.connect("localhost", 9090).sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        // 发送
        Channel channel = channelFuture.channel();
        byte[] requestBytes = IoUtil.serialize(content);
        DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_0, HttpMethod.POST, "/", Unpooled.copiedBuffer(requestBytes));
        request.headers().set(HttpHeaderNames.CONTENT_LENGTH, requestBytes.length);

        channel.writeAndFlush(request);
    }

    private void urlHttp(MyContent content, CompletableFuture<Object> completableFuture) {
        try {
            URL url = new URL("http://localhost:9090");
            HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();

            httpURLConnection.setRequestMethod("POST");
            httpURLConnection.setDoOutput(true);// 有body向外传输
            httpURLConnection.setDoInput(true);// 也要接收响应

            OutputStream os = httpURLConnection.getOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(os);
            oos.writeObject(content);// 这里并未真的发送

            if (httpURLConnection.getResponseCode() == HttpURLConnection.HTTP_OK) {// 这一步才真正发出去
                InputStream is = httpURLConnection.getInputStream();
                ObjectInputStream ois = new ObjectInputStream(is);
                MyContent respContent = (MyContent) ois.readObject();

                completableFuture.complete(respContent.getResponseObj());
            } else {
                System.out.println("url http 响应码" + httpURLConnection.getResponseCode());
            }
        } catch (MalformedURLException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }

    }


    public NioSocketChannel getClient (InetSocketAddress address) {
        ClientPool clientPool = oubBoxs.get(address);
        // System.out.println("首次获取连接池[key:"+address+"]结果："+clientPool);
        /*获取对象判空，要考虑并发情况，双重检查
        * 但此处这种写法类似双重检查：1.外层if；2.if absent
        * putIfAbsent jdk底层有加锁*/
        if (clientPool == null) {
            oubBoxs.putIfAbsent(address, new ClientPool(poolSize));
        }

        /*取出有效client*/
        clientPool = oubBoxs.get(address);
        // System.out.println("再次获取连接池[key:"+address+"]结果："+clientPool);
        int index = random.nextInt(poolSize);
        NioSocketChannel client = clientPool.clients[index];
        // System.out.println("首次获取连接池中位置【"+index+"】的连接："+client);
        if (client!=null && client.isActive()) {
            return client;
        } else {
            /*对当前client进行初始化 首次进入肯定是因为null*/
            synchronized (clientPool.locks[index]) {
                // 此处需要double check 否则连接池不管用
                client = clientPool.clients[index];
                // System.out.println("再次获取连接池中位置【"+index+"】的连接："+client);
                if (client!=null && client.isActive()) {
                    return client;
                }

                // System.out.println("创建连接放入位置【"+index+"】并返回");
                return client = clientPool.clients[index] = createClient (address);
            }
        }

        // 获取到了not active的client
        /*if (client!=null && !client.isActive()) {
            return getClient(address);
        }*/



    }

    private NioSocketChannel createClient (InetSocketAddress address) {
        // 基于netty的客户端
        clientWorker = new NioEventLoopGroup(1);
        Bootstrap bs = new Bootstrap();
        ChannelFuture connect = bs.group(clientWorker).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new PkgMsgDecoder());
                pipeline.addLast(new ClientResponseHandler());
            }
        }).connect(address);

        try {
            NioSocketChannel channel = (NioSocketChannel) connect.sync().channel();
            return channel;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
